unit Actors;

{$IFDEF FPC}
  {$MODE Delphi}
{$ENDIF}

interface

uses SysUtils, Classes, Variants, Windows, SyncObjs;

type

  TAction = procedure(const msg:Variant) of object;

  TItem = class(TObject)
  public
    Next: TItem;
  end;

  TQueue = class(TItem)
  private
    CS: TRTLCriticalSection;
    HasWork: TEvent;
    Head, Tail: TItem;
  public
    Count: Integer;
    constructor Create;
    destructor Destroy; override;
    procedure Push(w: TItem);
    function Pop(var w): Boolean;
  end;

  TActor = class(TQueue)
  private
    procedure Act;
  protected
    procedure Init(const msg:variant); virtual; // abstract;
  public
    procedure Send(a: TAction; const msg:Variant); overload;
  end;
  TActorClass = class of TActor;

  TWork = class(TItem)
    Action: TAction;
    Msg: Variant;
  end;

  TWorker = class(TThread)
  public
    procedure Execute; override;
  end;

  TThreadPool = class(TList)
    procedure AddThreads(n:Integer);
    procedure Remove(n:Integer);
    destructor Destroy; override;
  end;

  procedure Send(ac:TActorClass; const msg:Variant); overload;
  procedure Send(ac:TActorClass; const msg:array of Variant); overload;

var
  ActorsQueue: TQueue;
  Threads: TThreadPool;

implementation

procedure Send(ac:TActorClass; const msg:Variant);
var
  actor:TActor;
begin
  //todo: Create a cache of actors.
  actor := ac.Create;
  ActorsQueue.Push(actor);
  actor.Send(actor.Init, msg);
end;

procedure Send(ac:TActorClass; const msg:array of Variant);
begin
  Send(ac, VarArrayOf(msg));
end;

{ TActor }

procedure TActor.Init(const msg: variant);
begin

end;

procedure TActor.Act;
var
  w: TWork;
begin
  while Pop(w) do begin
    w.action(w.msg);
    w.Free;
  end;
end;

procedure TActor.Send(a:TAction; const msg:Variant);
var
  w: TWork;
begin
  w := TWork.Create;
  w.action := a;
  w.msg := msg;
  Push(w);
end;

//procedure TActor.Send(a:TAction; const msg:array of Variant);
//begin
//  Send(a, VarArrayOf(msg));
//end;

{ TQueue }

constructor TQueue.Create;
begin
  inherited Create;
  InitializeCriticalSection(CS);
  HasWork := TEvent.Create(nil, True, False, '');
end;

destructor TQueue.Destroy;
begin
  HasWork.Free;
  DeleteCriticalSection(CS);
end;

procedure TQueue.Push(w: TItem);
begin
  EnterCriticalSection(CS);
    if Head=nil then begin
      Head := w;
      Tail := w;
    end else begin
      Tail.next := w;
      Tail := w;
    end;
    Inc(Count);
  LeaveCriticalSection(CS);
  HasWork.SetEvent;
end;

function TQueue.Pop(var w): Boolean;
begin
  Result := HasWork.WaitFor(100) = wrSignaled;
  if Result then begin
    EnterCriticalSection(CS);
      Result := Count > 0;
      if Result then begin
        TItem(w) := Head;
        Head := Head.next;
        Dec(Count);
        if Count=0 then begin
          HasWork.ResetEvent;
          Tail := nil;
        end;
      end;
    LeaveCriticalSection(CS);
  end;
end;

{ TWorker }

procedure TWorker.Execute;
var
  a: TActor;
begin
  while not Terminated do
    // TActor(Queue.Pop).Act;
    while ActorsQueue.Pop(a) do
      a.Act;
end;

{ TThreadPool }

procedure TThreadPool.AddThreads(n: Integer);
var
  i: Integer;
begin
  for i := 1 to n do
    inherited Add(TWorker.Create(False));
end;

procedure TThreadPool.Remove(n: Integer);
var
  i: Integer;
begin
  for i := n-1 downto 0 do begin
    with TThread(Items[i]) do begin
      Terminate;
      WaitFor;
    end;
    Delete(i);
  end;
end;

destructor TThreadPool.Destroy;
begin
  Remove(Count);
  inherited;
end;

initialization
  ActorsQueue := TQueue.Create;

  Threads := TThreadPool.Create;
  Threads.AddThreads(4);

finalization
  ActorsQueue.Free;
  Threads.Free;
end.
